package com.stone.message.adapter.trigger;

import com.stone.message.adapter.domain.event.ReceiveMessageEvent;
import com.stone.message.adapter.infra.consumer.MessageConsumer;
import com.stone.message.adapter.infra.publisher.MessagePublisher;
import com.stone.message.adapter.listener.ReceiveMessageEventListener;
import com.stone.starter.core.exception.CommonException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Future;

/**
 * 公用抽象触发器（各种类型的触发器只用继承该类实现对应方法即可），该抽象类提供给具体的业务场景实现
 *
 * @param <T> 获取到的数据类型
 * @author Mr_wenpan@163.com 2021/07/14 21:55
 */
@Slf4j
public abstract class AbstractReceiverTrigger<T> {

    /**
     * 消息发送器
     */
    private final MessagePublisher messagePublisher;

    /**
     * 注入接受者消息监听器
     * todo 这里不能直接注入，ReceiveMessageEventListener靠条件注入，不是每个模块都必须要有消费者
     */
    @Autowired(required = false)
    private ReceiveMessageEventListener receiveMessageEventListener;

    /**
     * 停止标记
     */
    private volatile boolean stopFlag;

    public AbstractReceiverTrigger(MessagePublisher messagePublisher) {
        this.messagePublisher = messagePublisher;
        stopFlag = false;
    }

    /**
     * 触发执行方法
     *
     * @param queue     消息发送到哪个队列
     * @param threadNum 线程数量
     * @author Mr_wenpan@163.com 2021/7/15 1:49 下午
     */
    public void trigger(String queue, int threadNum) {

        if (!checkParam(queue, threadNum)) {
            log.warn("触发器校验未通过，请检查校验参数！queue={},threadNum={}", queue, threadNum);
            return;
        }

        // 是多线程异步执行还是同步执行由子类实现决定
        if (isAsyncExecute()) {
            List<Future<T>> futures = new ArrayList<>(threadNum);

            for (int i = 0; i < threadNum; i++) {
                futures.add(threadPoolTaskExecutor().submit(() -> publishMessage(queue)));
            }

            // 主线程等待
            for (Future<T> future : futures) {
                T result = null;
                try {
                    result = future.get();
                } catch (Exception ex) {
                    log.error("线程被抢占或中断，线程名称：[{}]，执行的数据是：{}，异常信息：{}",
                            Thread.currentThread().getName(), result, ex);
                }
            }

        } else {
            syncTriggerMessage(queue);
        }

    }

    protected T syncTriggerMessage(String queue) {
        // 由调用线程发送消息
        return publishMessage(queue);
    }

    private T publishMessage(String queue) {

        boolean isFaild = false;
        // 如果获取数据时出现异常，那么说明数据还没被获取到，不用做补偿
        T data = null;
        // 不断的获取数据，直到没有数据或调用了stop
        while (!stopFlag && Objects.nonNull(data = getData())) {
            try {
                // 将消息事件发布出去
                ReceiveMessageEvent<T> receiveMessageEvent = new ReceiveMessageEvent<>(this);
                receiveMessageEvent.setContent(data);
                receiveMessageEvent.setQueue(queue);
                messagePublisher.publish(receiveMessageEvent);
            } catch (Exception ex) {
                isFaild = true;
                log.error("任务执行出现异常，异常线程：[{}]，异常信息：{}", Thread.currentThread().getName(), ex);
            }
            if (isFaild) {
                isFaild = false;
                // 错误补偿
                errorCompensate(data);
            }
        }

        return data;
    }

    /**
     * 停止异步处理
     */
    public void stop() {
        stopFlag = true;
    }

    /**
     * 异步执行任务的线程池，由子类使用时动态传入
     *
     * @return org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
     * @author Mr_wenpan@163.com 2021/7/14 10:49 下午
     */
    protected abstract ThreadPoolTaskExecutor threadPoolTaskExecutor();

    /**
     * 获取待发送的数据，由子类决定如何获取数据
     *
     * @return T
     * @author Mr_wenpan@163.com 2021/7/14 10:49 下午
     */
    protected abstract T getData();

    /**
     * 任务执行失败的错误补偿，留给子类自己决定如何补偿，默认不做补偿
     *
     * @param data 处理的数据
     * @return T 处理的数据类型
     * @author Mr_wenpan@163.com 2021/7/14 10:48 下午
     */
    protected abstract T errorCompensate(T data);

    /**
     * 是否异步多线程触发执行，默认同步
     */
    protected boolean isAsyncExecute() {
        return Boolean.FALSE;
    }

    /**
     * 执行前的一些必要校验，留给子类自己决定如何校验
     */
    protected boolean check(String queue, int threadNum) {
        return Boolean.TRUE;
    }

    /**
     * 校验触发器必要参数
     *
     * @param queue     触发消息发送的队列
     * @param threadNum 线程数量（异步多线程执行才需要检查）
     * @author Mr_wenpan@163.com 2021/7/17 5:34 下午
     */
    private boolean checkParam(String queue, int threadNum) {
        if (StringUtils.isBlank(queue)) {
            return Boolean.FALSE;
        }
        if (isAsyncExecute()) {
            return threadNum > 0;
        }
        if (Objects.nonNull(receiveMessageEventListener)) {
            // 必须保证用户所指定的queue能正确获取到一个消费者来消费数据，避免拉取数据后没有消费者消费
            MessageConsumer consumer = receiveMessageEventListener.getConsumerByQueue(queue);
            if (Objects.isNull(consumer)) {
                throw new CommonException("指定的消息投递队列：[{}],找不到对应的消费者，请检查队列是否正确！");
            }
        }

        return Boolean.TRUE;
    }

}
